//! # 消息总线压力测试
//!
//! 包含极限场景、边界条件和高负载压力测试

use rustcloud_bus::prelude::*;
use rustcloud_bus::{
    bus::MessageBus,
    memory::{InMemoryMessageBus},
    message::{Message, MessagePriority},
    config::MessageBusConfig,
};
use serde_json::json;
use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
use std::time::{Duration, Instant};

/// 压力测试辅助工具
struct StressTestHelper;

impl StressTestHelper {
    /// 创建高负载配置的消息总线
    async fn create_high_load_bus() -> Arc<InMemoryMessageBus> {
        let config = MessageBusConfig {
            bus_type: rustcloud_bus::config::MessageBusType::Memory,
            max_connections: 10000,
            buffer_size: 50000,
            max_message_size: 10 * 1024 * 1024, // 10MB
            enable_metrics: true,
            ..Default::default()
        };
        
        Arc::new(InMemoryMessageBus::with_config(config).await.unwrap())
    }

    /// 创建大量测试消息
    fn create_messages(count: usize, topic_prefix: &str, payload_size: usize) -> Vec<Message> {
        (0..count).map(|i| {
            let topic = format!("{}.{}", topic_prefix, i % 100); // 100个不同主题
            let large_data = "x".repeat(payload_size);
            Message::new(&topic, json!({
                "id": i,
                "data": large_data,
                "timestamp": chrono::Utc::now().timestamp(),
                "sequence": i
            }))
        }).collect()
    }

    /// 并发发布消息
    async fn concurrent_publish(
        bus: Arc<InMemoryMessageBus>,
        messages: Vec<Message>,
        concurrency: usize
    ) -> (usize, Duration) {
        let start_time = Instant::now();
        let published_count = Arc::new(AtomicUsize::new(0));
        
        let chunk_size = messages.len() / concurrency;
        let mut handles = Vec::new();

        for chunk in messages.chunks(chunk_size) {
            let bus_clone = bus.clone();
            let published_count_clone = published_count.clone();
            let chunk_messages = chunk.to_vec();

            let handle = tokio::spawn(async move {
                for message in chunk_messages {
                    match bus_clone.publish(message).await {
                        Ok(()) => {
                            published_count_clone.fetch_add(1, Ordering::Relaxed);
                        }
                        Err(e) => {
                            eprintln!("发布失败: {}", e);
                        }
                    }
                }
            });
            handles.push(handle);
        }

        // 等待所有任务完成
        for handle in handles {
            handle.await.unwrap();
        }

        let duration = start_time.elapsed();
        let published = published_count.load(Ordering::Relaxed);
        
        (published, duration)
    }

    /// 并发订阅和消费消息
    async fn concurrent_consume(
        bus: Arc<InMemoryMessageBus>,
        subscriber_count: usize,
        pattern: &str,
        timeout_ms: u64
    ) -> Vec<usize> {
        let mut subscribers = Vec::new();
        let mut handles = Vec::new();

        // 创建订阅者
        for _ in 0..subscriber_count {
            let subscriber = bus.subscribe(pattern).await.unwrap();
            subscribers.push(subscriber);
        }

        // 启动消费任务
        for subscriber in subscribers {
            let handle = tokio::spawn(async move {
                let mut received_count = 0;
                let start_time = Instant::now();

                while start_time.elapsed().as_millis() < timeout_ms as u128 {
                    match subscriber.receive_timeout(100).await {
                        Ok(Some(_)) => {
                            received_count += 1;
                        }
                        Ok(None) => {
                            // 超时，继续
                        }
                        Err(_) => break,
                    }
                }
                received_count
            });
            handles.push(handle);
        }

        // 等待所有消费者完成
        let mut results = Vec::new();
        for handle in handles {
            let count = handle.await.unwrap();
            results.push(count);
        }

        results
    }
}

/// 极限场景测试模块
mod extreme_scenarios {
    use super::*;

    #[tokio::test]
    async fn test_massive_message_publishing() {
        println!("🚀 测试大量消息发布");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        let message_count = 100_000; // 10万条消息
        let payload_size = 1024; // 1KB per message
        
        let messages = StressTestHelper::create_messages(message_count, "stress.massive", payload_size);
        
        let start_time = Instant::now();
        let (published, duration) = StressTestHelper::concurrent_publish(bus.clone(), messages, 50).await;
        
        println!("📊 发布统计:");
        println!("  • 目标消息数: {}", message_count);
        println!("  • 实际发布数: {}", published);
        println!("  • 耗时: {:?}", duration);
        println!("  • 吞吐量: {:.2} msg/s", published as f64 / duration.as_secs_f64());
        
        // 验证结果
        assert!(published >= message_count * 95 / 100, "发布成功率应该 >= 95%");
        assert!(duration.as_secs() < 60, "应该在60秒内完成");
        
        // 验证统计信息
        let stats = bus.stats().await.unwrap();
        assert!(stats.total_published >= published as u64);
    }

    #[tokio::test]
    async fn test_high_concurrency_subscribers() {
        println!("👥 测试高并发订阅者");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        let subscriber_count = 1000; // 1000个订阅者
        
        // 创建大量订阅者
        let mut subscribers = Vec::new();
        for i in 0..subscriber_count {
            let pattern = format!("high.concurrency.{}", i % 10); // 10个不同模式
            let subscriber = bus.subscribe(&pattern).await.unwrap();
            subscribers.push(subscriber);
        }
        
        println!("📊 订阅者创建完成:");
        println!("  • 订阅者数量: {}", subscribers.len());
        
        // 发布测试消息
        for i in 0..10 {
            let topic = format!("high.concurrency.{}", i);
            let message = Message::new(&topic, json!({"test": i}));
            bus.publish(message).await.unwrap();
        }
        
        // 验证订阅者数量
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, subscriber_count);
        
        // 关闭订阅者
        for subscriber in subscribers {
            subscriber.close().await.unwrap();
        }
    }

    #[tokio::test]
    async fn test_memory_pressure_large_messages() {
        println!("💾 测试大消息内存压力");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        let message_count = 1000;
        let payload_size = 1024 * 1024; // 1MB per message = 1GB total
        
        let messages = StressTestHelper::create_messages(message_count, "stress.large", payload_size);
        
        println!("📊 大消息测试参数:");
        println!("  • 消息数量: {}", message_count);
        println!("  • 单条消息大小: {} MB", payload_size / (1024 * 1024));
        println!("  • 总数据量: {} MB", (message_count * payload_size) / (1024 * 1024));
        
        let start_time = Instant::now();
        let (published, duration) = StressTestHelper::concurrent_publish(bus.clone(), messages, 10).await;
        
        println!("📊 发布结果:");
        println!("  • 发布成功: {}", published);
        println!("  • 耗时: {:?}", duration);
        println!("  • 平均速度: {:.2} MB/s", 
                (published * payload_size) as f64 / (1024.0 * 1024.0 * duration.as_secs_f64()));
        
        assert!(published >= message_count * 80 / 100, "至少80%的大消息应该发布成功");
    }

    #[tokio::test]
    async fn test_topic_explosion() {
        println!("🎯 测试主题爆炸场景");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        let topic_count = 10_000; // 1万个不同主题
        
        // 为每个主题创建订阅者
        let mut subscribers = Vec::new();
        for i in 0..topic_count {
            let topic = format!("explosion.topic.{}", i);
            let subscriber = bus.subscribe(&topic).await.unwrap();
            subscribers.push(subscriber);
        }
        
        println!("📊 主题爆炸测试:");
        println!("  • 主题数量: {}", topic_count);
        println!("  • 订阅者数量: {}", subscribers.len());
        
        // 向每个主题发布消息
        let start_time = Instant::now();
        for i in 0..topic_count {
            let topic = format!("explosion.topic.{}", i);
            let message = Message::new(&topic, json!({"id": i}));
            bus.publish(message).await.unwrap();
        }
        let publish_duration = start_time.elapsed();
        
        println!("📊 发布完成:");
        println!("  • 发布耗时: {:?}", publish_duration);
        println!("  • 平均每主题: {:?}", publish_duration / topic_count as u32);
        
        // 验证主题列表
        let topics = bus.list_topics().await.unwrap();
        assert!(topics.len() >= topic_count, "应该创建了所有主题");
        
        // 清理订阅者
        for subscriber in subscribers {
            subscriber.close().await.unwrap();
        }
    }

    #[tokio::test]
    async fn test_rapid_subscribe_unsubscribe() {
        println!("🔄 测试快速订阅/取消订阅");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        let cycle_count = 1000; // 1000次循环
        
        let start_time = Instant::now();
        
        for i in 0..cycle_count {
            // 快速创建订阅者
            let subscriber = bus.subscribe(&format!("rapid.test.{}", i % 10)).await.unwrap();
            
            // 立即关闭
            subscriber.close().await.unwrap();
            
            if i % 100 == 0 {
                println!("  • 完成 {} 次循环", i);
            }
        }
        
        let duration = start_time.elapsed();
        
        println!("📊 快速订阅/取消订阅测试:");
        println!("  • 循环次数: {}", cycle_count);
        println!("  • 总耗时: {:?}", duration);
        println!("  • 平均每次: {:?}", duration / cycle_count as u32);
        
        // 验证最终状态
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 0, "所有订阅者都应该被清理");
    }

    #[tokio::test]
    async fn test_message_priority_under_load() {
        println!("⭐ 测试高负载下的消息优先级");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        let subscriber = bus.subscribe("priority.*").await.unwrap();
        
        // 发布不同优先级的消息
        let priorities = vec![
            MessagePriority::Low,
            MessagePriority::Normal, 
            MessagePriority::High,
            MessagePriority::Critical,
        ];
        
        let messages_per_priority = 1000;
        let mut published_count = 0;
        
        for priority in &priorities {
            for i in 0..messages_per_priority {
                let topic = format!("priority.{:?}", priority).to_lowercase();
                let mut message = Message::new(&topic, json!({
                    "priority": format!("{:?}", priority),
                    "id": i
                }));
                
                // 设置优先级
                let mut metadata = message.metadata().clone();
                metadata.attributes.priority = priority.clone();
                message = message.with_metadata(metadata);
                
                bus.publish(message).await.unwrap();
                published_count += 1;
            }
        }
        
        println!("📊 优先级测试:");
        println!("  • 总发布消息: {}", published_count);
        
        // 接收消息并检查顺序
        let mut received_messages = Vec::new();
        for _ in 0..100 { // 只检查前100条消息的顺序
            if let Some(message) = subscriber.receive_timeout(1000).await.unwrap() {
                let priority_str = message.payload()["priority"].as_str().unwrap();
                received_messages.push(priority_str.to_string());
            } else {
                break;
            }
        }
        
        println!("  • 接收到的消息: {}", received_messages.len());
        
        // 简单验证：关键消息应该在前面
        let critical_count = received_messages.iter()
            .take(20) // 前20条消息
            .filter(|p| p.contains("Critical"))
            .count();
        
        assert!(critical_count > 0, "前20条消息中应该有关键优先级消息");
    }
}

/// 边界条件测试模块
mod boundary_conditions {
    use super::*;

    #[tokio::test]
    async fn test_zero_buffer_size() {
        println!("🚫 测试零缓冲区大小");
        
        // 尝试创建零缓冲区的配置
        let config = MessageBusConfig {
            buffer_size: 0,
            ..Default::default()
        };
        
        // 这应该失败或使用默认值
        let result = InMemoryMessageBus::with_config(config).await;
        match result {
            Ok(bus) => {
                // 如果成功，应该使用了默认缓冲区大小
                let message = Message::new("test.zero.buffer", json!({"test": true}));
                let publish_result = bus.publish(message).await;
                assert!(publish_result.is_ok(), "即使缓冲区配置为0，也应该能发布消息");
            }
            Err(_) => {
                // 如果失败，这是预期的行为
                println!("  ✓ 零缓冲区配置被正确拒绝");
            }
        }
    }

    #[tokio::test]
    async fn test_maximum_message_size() {
        println!("📏 测试最大消息大小边界");
        
        let config = MessageBusConfig {
            max_message_size: 1024, // 1KB 限制
            ..Default::default()
        };
        
        let bus = InMemoryMessageBus::with_config(config).await.unwrap();
        
        // 测试刚好达到限制的消息
        let max_data = "x".repeat(1000); // 接近1KB
        let max_message = Message::new("test.max.size", json!({"data": max_data}));
        let result = bus.publish(max_message).await;
        assert!(result.is_ok(), "接近最大大小的消息应该发布成功");
        
        // 测试超过限制的消息
        let over_data = "x".repeat(2000); // 超过限制
        let over_message = Message::new("test.over.size", json!({"data": over_data}));
        let result = bus.publish(over_message).await;
        assert!(result.is_err(), "超过最大大小的消息应该被拒绝");
    }

    #[tokio::test]
    async fn test_empty_and_invalid_topics() {
        println!("🎯 测试空主题和无效主题");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        
        // 测试空主题
        let empty_topic_msg = Message::new("", json!({"test": "empty"}));
        let result = bus.publish(empty_topic_msg).await;
        assert!(result.is_err(), "空主题应该被拒绝");
        
        // 测试只有空格的主题
        let space_topic_msg = Message::new("   ", json!({"test": "spaces"}));
        let _result = bus.publish(space_topic_msg).await;
        // 这个可能成功也可能失败，取决于实现
        
        // 测试非常长的主题
        let long_topic = "a".repeat(10000);
        let long_topic_msg = Message::new(&long_topic, json!({"test": "long"}));
        let _result = bus.publish(long_topic_msg).await;
        // 长主题应该能处理
        
        // 测试特殊字符主题
        let special_chars = "test.特殊字符.🚀.symbols.#.$!@";
        let special_msg = Message::new(special_chars, json!({"test": "special"}));
        let result = bus.publish(special_msg).await;
        assert!(result.is_ok(), "特殊字符主题应该被支持");
    }

    #[tokio::test]
    async fn test_subscription_to_nonexistent_patterns() {
        println!("❓ 测试订阅不存在的模式");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        
        // 测试各种可能不存在的模式
        let patterns = vec![
            "nonexistent.topic",
            "*.nonexistent.*",
            "very.deep.nested.topic.that.does.not.exist",
            "special.chars.🦀.test",
            "numbers.123.456.789",
        ];
        
        for pattern in patterns {
            let result = bus.subscribe(pattern).await;
            assert!(result.is_ok(), "订阅任何模式都应该成功，即使主题不存在");
            
            let subscriber = result.unwrap();
            
            // 尝试接收消息（应该超时）
            let msg = subscriber.receive_timeout(100).await.unwrap();
            assert!(msg.is_none(), "不存在的主题不应该有消息");
            
            subscriber.close().await.unwrap();
        }
    }

    #[tokio::test] 
    async fn test_concurrent_access_same_topic() {
        println!("🔄 测试同一主题的并发访问");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        let topic = "concurrent.access.test";
        let concurrent_tasks = 100;
        
        // 启动多个并发发布者任务
        let mut publish_handles = Vec::new();
        
        // 发布者任务
        for i in 0..concurrent_tasks {
            let bus_clone = bus.clone();
            let topic_clone = topic.to_string();
            
            let handle = tokio::spawn(async move {
                for j in 0..10 {
                    let message = Message::new(&topic_clone, json!({
                        "publisher": i,
                        "sequence": j
                    }));
                    bus_clone.publish(message).await.unwrap();
                }
            });
            publish_handles.push(handle);
        }
        
        // 订阅者任务
        let mut subscribe_handles = Vec::new();
        for i in 0..concurrent_tasks {
            let bus_clone = bus.clone();
            let topic_clone = topic.to_string();
            
            let handle = tokio::spawn(async move {
                let subscriber = bus_clone.subscribe(&topic_clone).await.unwrap();
                let mut received = 0;
                
                for _ in 0..5 { // 每个订阅者尝试接收5条消息
                    if subscriber.receive_timeout(1000).await.unwrap().is_some() {
                        received += 1;
                    }
                }
                
                subscriber.close().await.unwrap();
                received
            });
            subscribe_handles.push(handle);
        }
        
        // 等待所有发布者任务完成
        for handle in publish_handles {
            handle.await.unwrap();
        }
        
        // 等待所有订阅者任务完成
        for handle in subscribe_handles {
            handle.await.unwrap();
        }
        
        // 验证最终状态
        let stats = bus.stats().await.unwrap();
        assert!(stats.total_published >= 1000, "应该发布了至少1000条消息");
        assert_eq!(stats.active_subscribers, 0, "所有订阅者都应该关闭");
    }
}

/// 资源限制测试模块
mod resource_limits {
    use super::*;

    #[tokio::test]
    async fn test_maximum_connections() {
        println!("🔗 测试最大连接数限制");
        
        let config = MessageBusConfig {
            max_connections: 100, // 限制100个连接
            ..Default::default()
        };
        
        let bus = InMemoryMessageBus::with_config(config).await.unwrap();
        let mut subscribers = Vec::new();
        
        // 尝试创建超过限制的订阅者
        for i in 0..150 {
            match bus.subscribe(&format!("connection.test.{}", i)).await {
                Ok(subscriber) => {
                    subscribers.push(subscriber);
                }
                Err(_) => {
                    println!("  ✓ 在第 {} 个连接处达到限制", i);
                    break;
                }
            }
        }
        
        println!("📊 连接测试结果:");
        println!("  • 实际创建的订阅者: {}", subscribers.len());
        
        // 应该接近但不超过限制
        assert!(subscribers.len() <= 100, "不应该超过最大连接数");
        
        // 清理
        for subscriber in subscribers {
            subscriber.close().await.unwrap();
        }
    }

    #[tokio::test]
    async fn test_buffer_overflow_handling() {
        println!("📦 测试缓冲区溢出处理");
        
        let config = MessageBusConfig {
            buffer_size: 10, // 很小的缓冲区
            ..Default::default()
        };
        
        let bus = InMemoryMessageBus::with_config(config).await.unwrap();
        let subscriber = bus.subscribe("overflow.test").await.unwrap();
        
        // 快速发布大量消息，不消费
        let mut published_count = 0;
        let mut failed_count = 0;
        
        for i in 0..100 {
            let message = Message::new("overflow.test", json!({"id": i}));
            match bus.publish(message).await {
                Ok(()) => published_count += 1,
                Err(_) => failed_count += 1,
            }
        }
        
        println!("📊 缓冲区溢出测试:");
        println!("  • 发布成功: {}", published_count);
        println!("  • 发布失败: {}", failed_count);
        
        // 由于缓冲区小，某些消息可能会失败
        assert!(published_count > 0, "至少应该发布一些消息");
        
        // 现在开始消费
        let mut consumed = 0;
        while let Some(_) = subscriber.receive_timeout(100).await.unwrap() {
            consumed += 1;
        }
        
        println!("  • 消费消息: {}", consumed);
        assert!(consumed > 0, "应该能消费一些消息");
        
        subscriber.close().await.unwrap();
    }

    #[tokio::test]
    async fn test_memory_usage_monitoring() {
        println!("📊 测试内存使用监控");
        
        let bus = StressTestHelper::create_high_load_bus().await;
        
        // 获取初始统计
        let initial_stats = bus.stats().await.unwrap();
        println!("📊 初始状态:");
        println!("  • 总消息数: {}", initial_stats.total_published);
        
        // 发布大量消息
        let message_count = 10000;
        for i in 0..message_count {
            let message = Message::new(
                &format!("memory.test.{}", i % 100),
                json!({
                    "id": i,
                    "data": "x".repeat(100), // 每条消息约100字节
                    "timestamp": chrono::Utc::now().timestamp()
                })
            );
            bus.publish(message).await.unwrap();
        }
        
        // 获取最终统计
        let final_stats = bus.stats().await.unwrap();
        println!("📊 最终状态:");
        println!("  • 总消息数: {}", final_stats.total_published);
        println!("  • 增加消息数: {}", final_stats.total_published - initial_stats.total_published);
        
        assert_eq!(
            final_stats.total_published - initial_stats.total_published,
            message_count as u64,
            "统计数字应该准确反映发布的消息数量"
        );
    }
}